feat(operator): static membership#346
Conversation
![cursor[bot]](https://avatars.githubusercontent.com/in/1210556?s=60&v=4){kind=small}
There was a problem hiding this comment.
Cursor Bugbot has reviewed your changes and found 1 potential issue.
❌ Bugbot Autofix is OFF. To automatically fix reported issues with cloud agents, enable autofix in the Cursor dashboard.
Reviewed by Cursor Bugbot for commit cfec667. Configure here.
| ) | ||
| def test_build_kafka_consumer_config( | ||
| source_config: StepConfig, | ||
| source_config: KafkaConsumerConfig, |
There was a problem hiding this comment.
Tests ignore static membership env
Low Severity
test_build_kafka_consumer_config asserts exact override_params but never clears STREAMS_KAFKA_GROUP_INSTANCE_ID. After this change, build_kafka_consumer_config injects group.instance.id from that env var, so these cases fail whenever it is set. The newer static-membership tests already use monkeypatch.delenv, so this looks like a missed update.
Reviewed by Cursor Bugbot for commit cfec667. Configure here.
| override_params = consumer_config.get("override_params", {}) | ||
| override_params = dict(consumer_config.get("override_params", {})) | ||
|
|
||
| group_instance_id = os.environ.get("STREAMS_KAFKA_GROUP_INSTANCE_ID") |
There was a problem hiding this comment.
You do not need to read environment variables from inside the adapter. The code that reads the config file is already able to transparently swap a config field with an environment variable provided according to a standard format
https://github.com/getsentry/streams/blob/main/sentry_streams/sentry_streams/pipeline/config.py#L26-L32
See an example of how this is used
- https://github.com/getsentry/streams/blob/main/sentry_streams/tests/pipeline/test_config.py#L45-L57
- https://github.com/getsentry/ops/blob/master/k8s/services/super-big-consumers/_items_log_config.yaml#L10
This means you can leverage the standard override mechanism without doing anything special for group.instance.id here.
More in general, beware of reading random environment variables deep in the stack. That makes it very hard to track down how to configure your system.
A better option is to have a config abstraction that provide the whole configuration and takes care to resolve, config files, overrides with environment variables and other ways to configure the system.
Then you separate the concerns cleanly. One system manages the loading of the config and another consumes it. The system consuming the config does not have to care where the value comes from.
| [ | ||
| pytest.param( | ||
| { | ||
| "starts_segment": None, |
There was a problem hiding this comment.
Why this ? start_segment is not an attribute of KafkaConsumerConfig
| def test_build_kafka_consumer_config_injects_static_membership( | ||
| monkeypatch: pytest.MonkeyPatch, | ||
| ) -> None: | ||
| monkeypatch.setenv("STREAMS_KAFKA_GROUP_INSTANCE_ID", "consumer-2") |
There was a problem hiding this comment.
You do not need to monkeypatch environment variables. Just set that with os.environ["blabla"] = "blabla".
In general please try to avoid monkey patching whenever it is not strictly necessary. It is really hard to makes sense of the behavior of the class you are patching when it evolves.
fpacifici
left a comment
fpacifici
left a comment
There was a problem hiding this comment.
Clicked the wrong button, please see my comments above
3 participants
Adds static membership to consumers (and also adds some type fix mypy was complaining about).